/* 
 *  Copyright (c) 2011—2013 Panguso.com. All rights reserved. 
 *  File Name:  Test.java                                      
 *  File Contents Outline :  并行日志预处理
 *  File  Create Date:   2012年7月30日
 *  File Version#: Revision#: #1                  
 *  File  Latest Modify Date : 2013年1月16日
 *  File Author :  bachuan@panguso.com
 */
package Test;

import java.io.ByteArrayOutputStream;
import java.io.ByteArrayInputStream;
import java.io.BufferedReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.regex.Pattern;
import cn.edu.thu.log.util.emails;
import cn.edu.thu.log.util.PrintClasspath;

import org.deckfour.xes.classification.XEventAttributeClassifier;
import org.deckfour.xes.extension.XExtension;
import org.deckfour.xes.extension.XExtensionManager;
import org.deckfour.xes.extension.std.XConceptExtension;
import org.deckfour.xes.extension.std.XTimeExtension;
import org.deckfour.xes.factory.XFactory;
import org.deckfour.xes.factory.XFactoryRegistry;
import org.deckfour.xes.info.XLogInfo;
import org.deckfour.xes.info.XLogInfoFactory;
import org.deckfour.xes.model.XAttribute;
import org.deckfour.xes.model.XAttributeMap;
import org.deckfour.xes.model.XEvent;
import org.deckfour.xes.model.XLog;
import org.deckfour.xes.model.XTrace;
import org.deckfour.xes.out.XesXmlSerializer;

import cn.edu.thu.log.clean.NoiseIdentify1;
import cn.edu.thu.log.filter.logfilter.LogFilter;
import cn.edu.thu.log.read.*;
import cn.edu.thu.log.util.Format;

import cn.edu.thu.log.util.PanGuConstants;
import cn.edu.thu.log.web.service.LogCleanConfigReadService;
import cn.edu.thu.log.web.service.WebConfigReadService;
import cn.edu.thu.log.web.service.impl.LogCleanConfigReadServiceImpl;
import cn.edu.thu.log.web.service.impl.WebConfigReadServiceImpl;
import cn.edu.thu.log.xes.XESConvertorMonitor;
import java.util.*;
import org.apache.hadoop.util.ToolRunner;
//import org.jdom.Document;
//import org.jdom.input.SAXBuilder;
//import org.jdom.output.XMLOutputter;
import storage.dbhelper.LogBodyHelper;
import storage.dbhelper.LogHeadHelper;
import storage.dbhelper.XESHelper;
import storage.dbobject.DbLogBody;
import storage.dbobject.DbLogHead;
import storage.dbobject.DbXES;
import java.util.Iterator;
import java.util.List;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.*;
import org.jdom.Document;
import org.jdom.input.SAXBuilder;
import org.jdom.output.XMLOutputter;
import cn.edu.thu.log.web.service.PropertiesReaderServices;
import cn.edu.thu.log.web.service.impl.PropertiesReaderServicesImpl;
import cn.edu.thu.log.util.ParameterConstructor;

// public class Test extends Configured implements Tool {
public class Test {

	static String param_pattern = "";
	static String param_version = "";
	static String param_inputPath = "";
	static String param_outputPath = "";
	static String param_xesName = "xes" + "_" + UUID.randomUUID();
	static String param_noiseFile = "";
	static String param_classify = "";
	static String param_startingdate = "";
	static String param_endingdate = "";
	static String param_productlist = "";
	static int clock = 0;
	static final String BRANCH = "-bn";
	static int numOfEvent = 0;
	int oldNumOfEvent = 0;
	static private int totalEvent = 0;
	static int XESBranch = 0;
	static String currentTimeString;
	public static XFactory factory;
	static XLog logReplace;
	static WebConfigReadService xesConfig;
	// static String filePath;
	String timestampTag;
	static String resultFilePath;
	static XLog log;
	// from logContent
	ArrayList<String> cateList;
	static LogConfig logConfig;
	static XEventAttributeClassifier classifier;
	XExtensionManager extensionManager;
	// for logBuffer clean and noise detect
	static LogBuffer testLogBuffer = new LogBuffer();
	// log format
	static String logHeadTokenizer;
	static String logBodyTokenizer;
	static String logHeadBodyTokenizer;
	// caseIDTagList
	static ArrayList<String> caseIDTagList;
	// LogHeadTag
	static ArrayList<String> logHeadTagList;
	// AvtivityList
	static ArrayList<String> activityTagList;
	static ArrayList<String> activityAbbreList;
	static ArrayList<String> activityLocList;
	static Hashtable<String, String> activityNameLocMap;
	// AttributeList
	static ArrayList<String> attributeTagList;
	static ArrayList<String> attributeAbbreList;
	static ArrayList<String> attributeLocList;
	static Hashtable<String, String> tagAbbreLocMap;
	static String classifyTags;
	// setupTimeStamp
	static String timeStampIndentifier;// dm
	static int timeStampLoc = 0;
	static ArrayList<Integer> caseIDLocList;
	static int totalEventNum = 0;
	private long timeTotal = 0;

	/**
	 * // * log Cleaner //
	 */
	// static LogClean logclean;
	/**
	 * noise Identifier
	 */
	static NoiseIdentify1 noiseidentify1;
	static LogCleanConfigReadService readLogCleanConfigService;
	static LogFilter logfilter1;

	/**
	 * Constructor
	 * 
	 * @param xesConfig
	 *            Reader to read configFile
	 * @param filePath
	 *            the file to save the result log
	 */
	public Test(WebConfigReadService xesConfig, String filePath) {
		factory = XFactoryRegistry.instance().currentDefault();
		// XFactoryRegistry.instance().setSizeOfShadowMap(10);
		Test.xesConfig = xesConfig;
		// this.filePath = filePath;
		logConfig = new LogConfig();
		// logclean = new LogClean();
		noiseidentify1 = new NoiseIdentify1();
		readLogCleanConfigService = LogCleanConfigReadServiceImpl.instance();
	}

	// initialize the config and logHead format
	static private void initConfig() {
		logConfig.config();
		logHeadTokenizer = logConfig.getLogHeadTokenizer();
		logBodyTokenizer = logConfig.getLogBodyTokenizer();
		logHeadBodyTokenizer = logConfig.getLogHeadBodyTokenizer();
		// setUpcaseIDList
		caseIDTagList = xesConfig.getPattern().getCaseIDList();// dm
		// mainCase
		String mainCase = xesConfig.getPattern().getMainCaseID();// dm
		if (caseIDTagList.size() == 0) {
			caseIDTagList.add(mainCase);
		}
		// setAbbreLogMap
		tagAbbreLocMap = new Hashtable<String, String>();
		// setLogHeadTag
		logHeadTagList = xesConfig.getVersion().getLogHeadList();// dm
		// setAvtivityList
		activityTagList = xesConfig.getPattern().getActivityList();// dm
		activityAbbreList = new ArrayList<String>();
		activityLocList = new ArrayList<String>();
		activityNameLocMap = new Hashtable<String, String>();
		setActivityAbbre(activityTagList, activityAbbreList, activityLocList);
		// setup classiferTags
		classifyTags = xesConfig.getPattern().getClassifytag();
		// test for nulkl
		noiseidentify1 = new NoiseIdentify1();
		readLogCleanConfigService = LogCleanConfigReadServiceImpl.instance();
		// if(classifyTags!=null)
		readLogCleanConfigService.setClassifyTags(classifyTags);

		// setAttributeList
		attributeTagList = xesConfig.getPattern().getAttributeList();// dm
		attributeAbbreList = new ArrayList<String>();
		attributeLocList = new ArrayList<String>();
		setActivityAbbre(attributeTagList, attributeAbbreList, attributeLocList);
		// System.out.println("attributeTagList:" + attributeTagList);
		// System.out.println("attributeAbbreList:" + attributeAbbreList);
		// System.out.println("attributeLocList:" + attributeLocList);

		// setupTimeStamp and timeStamp location
		timeStampIndentifier = xesConfig.getPattern().getTimeStamp();// dm
		for (int j = 0; j < logHeadTagList.size(); j++) {
			if (timeStampIndentifier.equalsIgnoreCase(logHeadTagList.get(j))) {
				timeStampLoc = getLogHeadLoc(j);
			}
		}

		// setup caseID and caseID location
		caseIDLocList = new ArrayList<Integer>();
		for (int i = 0; i < caseIDTagList.size(); i++) {
			String caseID = caseIDTagList.get(i);
			for (int j = 0; j < logHeadTagList.size(); j++) {
				if (caseID.equalsIgnoreCase(logHeadTagList.get(j))) {
					int loc = getLogHeadLoc(j);
					caseIDLocList.add(loc);

				}
			}
		}
	}

	/**
	 * set up classifier for log
	 */
	static private void setupClassifier(XLog log) {
		String[] keys = { XConceptExtension.KEY_NAME };
		classifier = new XEventAttributeClassifier("activity classifier", keys);
		log.getClassifiers().add(classifier);

	}

	// set up extensions for log
	static private void setupExtensions(XLog log) {
		XExtension timeExtension = XTimeExtension.instance();
		XExtension conceptExtension = XConceptExtension.instance();
		log.getExtensions().add(timeExtension);
		log.getExtensions().add(conceptExtension);

	}

	// set up global attributes
	static private void setupGlobalAttributes(XLog log) {
		factory = XFactoryRegistry.instance().currentDefault();
		XAttribute attributeConcept = factory.createAttributeLiteral(
				XConceptExtension.KEY_NAME, "name", null);
		log.getGlobalEventAttributes().add(attributeConcept);
		XAttribute attributeTime = factory.createAttributeLiteral(
				XTimeExtension.KEY_TIMESTAMP, "time", null);
		log.getGlobalEventAttributes().add(attributeTime);

		log.getGlobalTraceAttributes().add(attributeConcept);
	}

	/**
	 * get teh location of content in logHead
	 * 
	 * @param tagLocation
	 *            the location of the tag in tagList
	 * @return the location of content in contentList in logHead
	 */
	static private int getLogHeadLoc(int tagLocation) {
		LogHeadHelper logHeadH = new LogHeadHelper();
		List<DbLogHead> logHeadTags = logHeadH.getLogHeads(DbLogHead.cl_name,
				logHeadTagList.get(tagLocation));
		DbLogHead logHeadTag = logHeadTags.get(0);
		String locString = logHeadTag.getLocation();

		int loc = Integer.parseInt(locString) - 1;
		return loc;
	}

	/**
	 * set up the activity tag in each logBuffer
	 * 
	 * @param activityIDTagList
	 *            the activities' tags eat up by user
	 * @param logTagList
	 *            the tag list read from config reader
	 * @return the list of location of activityTag in whole tagList
	 */
	static private void setActivityAbbre(ArrayList<String> actvityList,
			ArrayList<String> activityAbbreList,
			ArrayList<String> activityLocList) {

		LogBodyHelper logBodyH = new LogBodyHelper();
		LogHeadHelper logHeadH = new LogHeadHelper();
		// System.out.println("important:actvityList size:" + actvityList.size()
		// + "\n" + actvityList);
		for (int i = 0; i < actvityList.size(); i++) {
			String activityName = actvityList.get(i);
			// System.out.println("important:activityName:" + activityName);
			List<DbLogBody> logBodys = logBodyH.getLogBodys(DbLogBody.cl_name,
					activityName);

			if (logBodys.size() != 0) {// avtivity belong to logBody
				DbLogBody logBody = logBodys.get(0);
				String abbre = logBody.getAbbreviation();
				activityAbbreList.add(abbre);
				if (!tagAbbreLocMap.contains(abbre)) {// add to location and
														// abbre map
					tagAbbreLocMap.put(abbre, activityName);
				}
			} else {// avtivity belong to logHead
				List<DbLogHead> logHeads = logHeadH.getLogHeads(
						DbLogHead.cl_name, activityName);
				if (logHeads.size() != 0) {
					DbLogHead logHead = logHeads.get(0);
					String location = logHead.getLocation();
					activityLocList.add(location);
					if (!tagAbbreLocMap.contains(location)) {// add to location
																// and name map
						tagAbbreLocMap.put(location, activityName);
						if (!activityNameLocMap.containsKey(location)) {
							activityNameLocMap.put(location, activityName);
						}
						// System.out.println("tagAbbreLocMap.put location:" +
						// location+" activityName:"+activityName);
					}
				}
			}
		}
		return;
	}

	/**
	 * write activities,logPath,caseID to event
	 * 
	 * @param event
	 * @param logBuffer
	 */
	static void writeEvent(XEvent event, LogBuffer logBuffer) {

		XFactory factory = XFactoryRegistry.instance().currentDefault();
		ArrayList<Object> logActivityContents = logBuffer
				.getActivityIDContentList();
		XAttributeMap attributeMap = factory.createAttributeMap();
		String activityVal = "";
		for (int i = 0; i < logActivityContents.size(); i++) {
			if (logActivityContents.get(i) != null) {
				String logActivityL = logActivityContents.get(i).toString();// ww
				if (i == 0) {
					activityVal = logActivityL;
				} else {
					activityVal = activityVal + "," + logActivityL;
				}

			}
		}
		if (activityVal.length() > 0) {
			XAttribute attribute = factory.createAttributeLiteral(
					XConceptExtension.KEY_NAME, activityVal, null);
			attributeMap.put(XConceptExtension.KEY_NAME, attribute);

		} else {
			XAttribute attribute = factory.createAttributeLiteral(
					XConceptExtension.KEY_NAME, "NA", null);
			attributeMap.put(XConceptExtension.KEY_NAME, attribute);
		}

		// get attribute of this event
		Hashtable<String, String> logAttibuteContentMap = logBuffer
				.getAttributeIDContentMap();
		Enumeration keyEnumer = logAttibuteContentMap.keys();
		// put the log tags as attributes to attributeMap

		while (keyEnumer.hasMoreElements()) {
			String attributeTagName = keyEnumer.nextElement().toString();
			String attributeContent = logAttibuteContentMap
					.get(attributeTagName);
			if (attributeContent != null && attributeContent.length() > 0) {
				XAttribute attribute = factory.createAttributeLiteral(
						attributeTagName, attributeContent, null);
				attributeMap.put(attributeTagName, attribute);
			} else {// content==null or content=""
				XAttribute attribute = factory.createAttributeLiteral(
						attributeTagName, "NA", null);
				attributeMap.put(attributeTagName, attribute);
			}

		}

		String timeString = logBuffer.getTimeStamp();
		Date timeStamp = Format.StringToTimeStamp(timeString);
		XAttribute attributeTime = factory.createAttributeTimestamp(
				XTimeExtension.KEY_TIMESTAMP, timeStamp, null);
		attributeMap.put(attributeTime.getKey(), attributeTime);
		currentTimeString = timeString;
		// set event's AttributesMap
		event.setAttributes(attributeMap);
	}

	static public void splitFinalLogInReduce(String name, int branch) {
		factory = XFactoryRegistry.instance().currentDefault();
		while (numOfEvent > PanGuConstants.MAX_EVENT_NUM) {
			// System.out.println("numOfEvent:"+numOfEvent);
			Iterator<XTrace> iterator = log.iterator();
			List<XTrace> tempXTraceList = new ArrayList<XTrace>();
			while (iterator.hasNext()) {
				XTrace currentTrace = iterator.next();
				logReplace.add(currentTrace);
				XESConvertorMonitor.instance().numOfEvent = XESConvertorMonitor
						.instance().numOfEvent + currentTrace.size();
				tempXTraceList.add(currentTrace);
				if (XESConvertorMonitor.instance().numOfEvent > PanGuConstants.MAX_EVENT_NUM) {
					numOfEvent = numOfEvent - currentTrace.size();
					writeToXES(logReplace, name, branch);
					logReplace.clear();
					logReplace = factory.createLog();
					XESConvertorMonitor.instance().numOfEvent = 0;
					break;
				} else {
					numOfEvent = numOfEvent - currentTrace.size();
				}
			}
			log.removeAll(tempXTraceList);
		}
	}

	static private boolean splitXLogInReduce(String name, int branch) {
		int count = 0;
		factory = XFactoryRegistry.instance().currentDefault();
		if (numOfEvent > PanGuConstants.MUTIPLE_MAX_EVENT_NUM
				* PanGuConstants.MAX_EVENT_NUM) {
			Iterator<XTrace> iterator = log.iterator();
			List<XTrace> tempXTraceList = new ArrayList<XTrace>();
			while (iterator.hasNext()) {
				XTrace currentTrace = iterator.next();
				logReplace.add(currentTrace);
				XESConvertorMonitor.instance().numOfEvent = XESConvertorMonitor
						.instance().numOfEvent + currentTrace.size();
				tempXTraceList.add(currentTrace);

				if (XESConvertorMonitor.instance().numOfEvent > PanGuConstants.MAX_EVENT_NUM) {
					numOfEvent = numOfEvent - currentTrace.size();
					writeToXES(logReplace, name, branch);
					logReplace.clear();
					logReplace = factory.createLog();
					XESConvertorMonitor.instance().numOfEvent = 0;
					break;
				} else {
					numOfEvent = numOfEvent - currentTrace.size();
				}
			}
			log.removeAll(tempXTraceList);
			// 728 for delete
			if (!iterator.hasNext()) 
				return false;			
		}
		return true;
	}

	static private void writeToXES(XLog writeLog, String name, int branch) {
		// wirteLog is empty
		if (writeLog.size() == 0) {
			return;
		}
		try {

			XesXmlSerializer xstream = new XesXmlSerializer();

			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			xstream.serialize(writeLog, baos);
			SAXBuilder builder = new SAXBuilder();
			Document doc;

			XMLOutputter outputter = new XMLOutputter();
			ByteArrayInputStream bais = new ByteArrayInputStream(
					baos.toByteArray());
			doc = builder.build(bais);
			String xesString = outputter.outputString(doc);

			XESHelper xesh = new XESHelper();
			Date currentTime = new Date();
			// String xesString = URLEncoder.encode(xesStringTemp, "UTF-8");
			String branchWithUUID = Integer.toString(XESBranch) + "_"
					+ UUID.randomUUID();
			DbXES dbxes = new DbXES(branchWithUUID, name, xesString,
					currentTime.toString(), param_classify);
			// System.out.println("write xes name:" + name + " branch:" +
			// branchWithUUID);
			xesh.saveXES(dbxes);
			baos.close();
			bais.close();

			XESBranch++;
			try {
				// sFile.delete();//dm
			} catch (Exception e) {
			}

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public LogBuffer getTestLogBuffer() {
		return testLogBuffer;
	}

	static public void setTestLogBuffer(LogBuffer logBuffer) {
		Test.testLogBuffer = logBuffer;
	}

	public XLog getLog() {
		return log;
	}

	public void setLog(XLog log) {
		Test.log = log;
	}

	private int getLogSize(XLog logL) {
		int events = 0;
		XLogInfo info = null;
		info = XLogInfoFactory.createLogInfo(logL);
		events += info.getNumberOfEvents();
		return events;
	}

	/**
	 * @return the totalEvent
	 */
	static public int getTotalEvent() {
		return totalEvent;
	}

	/**
	 * @param totalEvent
	 *            the totalEvent to set
	 */
	static public void setTotalEvent(int totalEvent) {
		Test.totalEvent = totalEvent;
	}

	/**
	 * @return the timeTotal
	 */
	public long getTimeTotal() {
		return timeTotal;
	}

	/**
	 * @param timeTotal
	 *            the timeTotal to set
	 */
	public void setTimeTotal(long timeTotal) {
		this.timeTotal = timeTotal;
	}

	static int printUsage() {
		// System.out.println("log [-m <maps>] [-r <reduces>] <input> <output>");
		ToolRunner.printGenericCommandUsage(System.out);
		return -1;
	}

	public static class Map extends Mapper<LongWritable, Text, Text, Text> {

		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		// private int timecount1 = 0 ;

		// private Text eventText = new Text();
		// private EventBufferWritable eventBufferWritable;
		@Override
		public void cleanup(Mapper.Context context) {
			// System.out.println("timer1:"+Timer.formatDuration(timecounter1));
			// System.out.println("timer2:"+Timer.formatDuration(timecounter2));
			// System.out.println("timer3:"+Timer.formatDuration(timecounter3));
			// System.out.println("timer4:"+Timer.formatDuration(timecounter4));
			// System.out.println("timer2:"+Timer.formatDuration(timecounter2));
		};

		@Override
		public void setup(Context context) throws IOException,
				InterruptedException {
			System.out.println("map begin");
			PrintClasspath.print();
			// compiled code
			// Timer t1 = new Timer();
			// filePath =
			// "E:/盘古日志0419/log3/collect_temp_3/page/pagesearch/10.10.128.13/pagesearch_201203010000_0.log";
			// t1.start();
			logConfig = new LogConfig();

			// System.out.println("initial LogConfig time elapse:"+t1.getDurationString());

			WebConfigReadService configServiceL = WebConfigReadServiceImpl
					.instance();

			Configuration cnfiguration = context.getConfiguration();
			param_pattern = cnfiguration.get("param_pattern");
			param_version = cnfiguration.get("param_version");
			configServiceL.setPatternName(param_pattern);
			// System.out.println("map param_pattern:" + param_pattern);
			configServiceL.setVersionName(param_version);
			// System.out.println("map param_version:" + param_version);
			// t1.start();
			configServiceL.config();
			// t1.stop();
			// System.out.println("configservice time elapse:"+t1.getDurationString());
			xesConfig = configServiceL;
			initConfig();
			param_noiseFile = cnfiguration.get("param_noiseFile");
			// t1.start();

			if (param_noiseFile != null) {
				if (param_noiseFile.length() != 0) {
					readLogCleanConfigService.setNoiseflag(1);// 说明已经选择noise文件
					readLogCleanConfigService.setNoiseFileName(param_noiseFile);
					noiseidentify1
							.readNoiseIdetifyConfigFile(readLogCleanConfigService);
					System.out.println("noise file:" + param_noiseFile);
				}
			}
			logfilter1 = new LogFilter();
			logfilter1.setClassifyTags(classifyTags);
			// t1.stop();
			// System.out.println("map param_noiseFile:" +
			// param_noiseFile+" and config time elapse:"+t1.getDurationString());
		}

		// 每次执行map都会自动执行configure函数，可以通过这个函数给变量赋值

		// static {
		// factory = XFactoryRegistry.instance().currentDefault();
		// filePath =
		// "E:/盘古日志0419/log3/collect_temp_3/page/pagesearch/10.10.128.13/pagesearch_201203010000_0.log";
		// logConfig = new LogConfig();
		// WebConfigReadService configServiceL =
		// WebConfigReadServiceImpl.instance();
		// // configServiceL.setPatternName("page_HasAds");
		// // configServiceL.setPatternName("ContentType_attributes");
		// // configServiceL.setVersionName("version1.5");
		// configServiceL.setPatternName(param_pattern);
		// System.out.println("map param_pattern:" + param_pattern);
		// configServiceL.setVersionName(param_version);
		// System.out.println("map param_version:" + param_version);
		// configServiceL.config();
		// xesConfig = configServiceL;
		// initConfig();
		// }

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			// XESConvertorMonitor.instance().numOfEvent = 0;
			String mapOutput = "";
			// Timer TempTimer = new Timer();
			// TempTimer.start();
			// init
			String logHeadContent = "";
			String logBodyContent = "";
			LogBuffer logBuffer;
			// XEvent event;// XES event
			BufferedReader reader; // read each log Record
			String record;
			// ------------------witeToEvent-----------------
			// initialize
			// event = factory.createEvent();
			logBuffer = new LogBuffer();
			String temprecord = value.toString();
			ArrayList<Object> params = new ArrayList<Object>();
			ArrayList<Object> headparams = new ArrayList<Object>();
			ArrayList<Object> bodyparams = new ArrayList<Object>();
			Hashtable<String, String> bodyMap;
			// split log head and log body
			String patternString = "[^\\\\]";
			patternString = patternString.concat(logHeadBodyTokenizer);
			Pattern pattern = Pattern.compile(patternString);
			String[] strs = pattern.split(temprecord);
			if (strs[0].length() == temprecord.length()) {
			}
			if (strs[0].length() == 0) {// 日志头为空
				word = new Text(PanGuConstants.MR_REMOVE);
				context.write(word, new Text("1"));
				return;
			}

			logHeadContent = strs[0].concat(String.valueOf(temprecord
					.charAt(strs[0].length() - 1)));
			if (strs.length > 1) {
				logBodyContent = strs[1];
			}
			// set up log Head params
			headparams.addAll(Format.splitContentToList(logHeadTokenizer,
					logHeadContent));
			// set up logBody params
			bodyparams.addAll(Format.splitContentToList(logBodyTokenizer,
					logBodyContent));
			bodyMap = Format
					.splitContentToMap(logBodyTokenizer, logBodyContent);
			// set up log Content
			params.addAll(headparams);
			params.addAll(bodyparams);

			/*
			 * ---set up logBuffer and put it into event-------
			 */
			logBuffer.setLogBodyMap(bodyMap);
			// logBuffer.setLogPath(filePath);
			logBuffer.setLogHeadContent(headparams);
			logBuffer.setLogBodyContent(bodyparams);
			logBuffer.setLogBody(logBodyContent);
			logBuffer.setLogContent(params);
			logBuffer.setLogHeadTagList(logHeadTagList);// dm

			// for noise and clean
			setTestLogBuffer(logBuffer);
			/**
			 * 首先扫描判断记录的日志头是否与要求的一致
			 */
			if (logBuffer.getLogHeadTagList().size() != logBuffer
					.getLogHeadContent().size()) {
				word = new Text(PanGuConstants.MR_REMOVE);
				context.write(word, new Text("1"));
				return;
			}
			// 活动抽象,如果活动列表中包括这个字段，则转换为xes时，修改logbuffer，否则不修改
			// 由于是单个字段修改，而不是组合字段，则直接在日志清洗的时候进行

			// 日志清洗、噪声识别,在清洗中实现分段规则

			if (readLogCleanConfigService.getNoiseflag() == 1) {
				// Timer noiseTimer = new Timer();
				// noiseTimer.start();
				if (noiseidentify1.logNoise(logBuffer)) {
					word = new Text(PanGuConstants.MR_REMOVE);
					context.write(word, new Text("1"));
					return;
				}

			} // 如果没有设置清洗规则，但是设定了分段
			else if (!readLogCleanConfigService.getClassifyTags().equals("")) {
				logfilter1.filter(logBuffer);
			}
			String timeStampContent = logBuffer.getLogHeadContent()
					.get(timeStampLoc).toString();
			logBuffer.setTimeStamp(timeStampContent);

			// set caseID tag and content to LogBuffer
			ArrayList<String> caseIDContentList = new ArrayList<String>();
			String logCaseIDString = new String("");
			for (int i = 0; i < caseIDLocList.size(); i++) {
				int caseIDLoc = caseIDLocList.get(i);
				if (caseIDLoc <= logBuffer.getLogHeadContent().size()) {

					String caseIDContent = logBuffer.getLogHeadContent()
							.get(caseIDLoc).toString();
					// System.out.println(" logBuffer.getLogHeadContent():"+
					// logBuffer.getLogHeadContent());
					// System.out.println("caseIDLoc"+ caseIDLoc);
					// System.out.println("caseIDContent:"+caseIDContent);
					caseIDContentList.add(caseIDContent);
					logCaseIDString += caseIDContent + ",";
				} else {
					// caseIDContentList.add("NA");
					logCaseIDString += "NA" + ",";
					// System.out.println("activity not in logHead");
				}

			}
			// if caseID does not exit, discard this record
			if (logCaseIDString.contains("NA,")) {
				// System.out.println("logCaseIDString.equalsIgnoreCase NA");
				word = new Text(PanGuConstants.MR_REMOVE);
				context.write(word, new Text("1"));
				return;
			}
			if (logCaseIDString.equals(",")) {
				// System.out.println("logCaseIDString has no userID,delete");
				word = new Text(PanGuConstants.MR_REMOVE);
				context.write(word, new Text("1"));
				return;
			}
			// logBuffer.setCaseIDList(caseIDContentList);//do not need anymore
			logBuffer.setCaseIDString(logCaseIDString);

			// set activityContent to LogBuffer
			mapOutput += PanGuConstants.MR_ACTIVITY;
			ArrayList<Object> activityIDContentList = new ArrayList<Object>();
			for (int l = 0; l < activityAbbreList.size(); l++) {
				String activityAbbre = activityAbbreList.get(l);
				activityIDContentList.add(activityAbbre + ":"
						+ bodyMap.get(activityAbbre));
				mapOutput += activityAbbre + ":" + bodyMap.get(activityAbbre)
						+ PanGuConstants.MR_ACTIVITY_TOKEN;
				;// add to mapoutput
				// int startIndex = logBuffer.getLogBody().indexOf(activityAbbre
				// + ":");
				// if (startIndex >= 0) {
				// String subString =
				// logBuffer.getLogBody().substring(startIndex);
				// int endIndex = subString.indexOf(logBodyTokenizer);
				// String activityContent = subString.substring(0, endIndex);
				//
				// }
			}
			for (int l = 0; l < activityLocList.size(); l++) {
				String activityLoc = activityLocList.get(l);
				int loc = Integer.parseInt(activityLoc) - 1;
				if (loc <= logBuffer.getLogHeadContent().size()) {
					Object activityL = logBuffer.getLogHeadContent().get(loc);
					if (activityL != null) {
						if (activityL.toString().length() > 0) {
							activityIDContentList.add(activityNameLocMap
									.get(activityLoc) + ":" + activityL);
							mapOutput += activityNameLocMap.get(activityLoc)
									+ ":" + activityL
									+ PanGuConstants.MR_ACTIVITY_TOKEN;
							;// add to mapoutput

						} else {
							activityIDContentList.add("NA");
							mapOutput += "NA"
									+ PanGuConstants.MR_ACTIVITY_TOKEN;
							;// add to mapoutput
						}
					}
				} else {
					activityIDContentList.add("NA");
					mapOutput += "NA" + PanGuConstants.MR_ACTIVITY_TOKEN;
					;// add to mapoutput
					// System.out.println("activity not in logHead");
				}
			}
			mapOutput += PanGuConstants.MR_TOKEN;// end of activity in mapoutput
			// have not found match avtivity
			if (activityIDContentList.size() == 0) {
				// System.out.println("activityIDContentList.size() == 0");
				word = new Text(PanGuConstants.MR_REMOVE);
				context.write(word, new Text("1"));

				return;
			}

			// logBuffer.setActivityIDContentList(activityIDContentList);//do
			// not need anymore

			// mapoutput add activity
			// mapOutput = mapOutput + PanGuConstants.MR_ACTIVITY;
			// for (int i = 0; i < activityIDContentList.size(); i++) {
			// mapOutput += activityIDContentList.get(i).toString() +
			// PanGuConstants.MR_ACTIVITY_TOKEN;
			// }
			// mapOutput += PanGuConstants.MR_TOKEN;

			// set attributeLiat to LogBuffer
			mapOutput += PanGuConstants.MR_ATTRIBUTE;// start of attribute in
														// mapOutput
			Hashtable<String, String> attributeIDContenMap = new Hashtable<String, String>();// <TagName,TagContent>
			for (int l = 0; l < attributeAbbreList.size(); l++) {
				String attributeAbbre = attributeAbbreList.get(l);
				String tagName = tagAbbreLocMap.get(attributeAbbre);
				String bodyContent = bodyMap.get(attributeAbbre);
				if (bodyContent != null) {
					if (bodyContent.toString().length() > 0) {
						attributeIDContenMap.put(tagName, bodyContent);
						mapOutput += tagName + ":" + bodyContent
								+ PanGuConstants.MR_ATTRIBUTE_TOKEN;// add to
																	// mapOutput
					} else {
						attributeIDContenMap.put(tagName, PanGuConstants.NA);// add
																				// to
																				// mapOutput
						mapOutput += tagName + ":" + PanGuConstants.NA
								+ PanGuConstants.MR_ATTRIBUTE_TOKEN;// add to
																	// mapOutput
					}
					// attributeIDContenMap.put(tagName, bodyContent);
				} else {
					attributeIDContenMap.put(tagName, "NA");
					mapOutput += tagName + ":" + PanGuConstants.NA
							+ PanGuConstants.MR_ATTRIBUTE_TOKEN;// add to
																// mapOutput
				}
			}
			for (int l = 0; l < attributeLocList.size(); l++) {
				String attributeLoc = attributeLocList.get(l);
				int loc = Integer.parseInt(attributeLoc);
				if (loc < logBuffer.getLogHeadContent().size()) {
					String tagName = tagAbbreLocMap.get(String.valueOf(loc));
					Object headContent = logBuffer.getLogHeadContent().get(
							loc - 1);
					// Object attributeL =
					// logBuffer.getLogHeadContent().get(loc);
					if (headContent != null) {
						if (tagName == null) {
							// System.out.println("(tagName==null:loc:" + loc);
						}
						if (headContent.toString().length() > 0) {
							attributeIDContenMap.put(tagName,
									headContent.toString());
							mapOutput += tagName + ":" + headContent.toString()
									+ PanGuConstants.MR_ATTRIBUTE_TOKEN;// add
																		// to
																		// mapOutput
						} else {
							attributeIDContenMap.put(tagName, "NA");
							mapOutput += tagName + ":" + PanGuConstants.NA
									+ PanGuConstants.MR_ATTRIBUTE_TOKEN;// add
																		// to
																		// mapOutput
						}
						// attributeIDContenMap.put(tagName,
						// headContent.toString());
					} else {
						attributeIDContenMap.put(tagName, "NA");
						mapOutput += tagName + ":" + PanGuConstants.NA
								+ PanGuConstants.MR_ATTRIBUTE_TOKEN;// add to
																	// mapOutput
					}
				}
			}
			mapOutput += PanGuConstants.MR_TOKEN;// end of attribute in
													// mapOutput
			// logBuffer.setAttributeIDContentMap(attributeIDContenMap);//do not
			// need anymore

			// mapoutput add attribute
			// mapOutput += mapOutput + PanGuConstants.MR_ATTRIBUTE;
			// Enumeration attriEnum = attributeIDContenMap.keys();
			// while (attriEnum.hasMoreElements()) {
			// String attriKey = attriEnum.nextElement().toString();
			// mapOutput += attriKey + ":" + attributeIDContenMap.get(attriKey)
			// + PanGuConstants.MR_ATTRIBUTE_TOKEN;
			// }
			// mapOutput += PanGuConstants.MR_TOKEN;
			// mapoutput add logpath
			// mapOutput = mapOutput + PanGuConstants.MR_FILEPATH + filePath +
			// PanGuConstants.MR_TOKEN;

			// map output
			// TempTimer2.stop();
			// timecounter4 += TempTimer2.getDuration();
			// System.out.println("日志后处理时间："+TempTimer2.getDurationString());
			word = new Text(logCaseIDString + timeStampContent);
			context.write(word, new Text(mapOutput));

		}
	}

	/**
	 * A reducer class that just emits the sum of the input values.
	 */
	public static class Reduce extends Reducer<Text, Text, Text, Text> {
		// String name = "mapreduce";

		// String name = param_xesName;
		String preCaseID = "";
		String preSummary = "";
		int preEventNum = 0;
		int preBranch = -1;
		String preTimestamp = "";
		static XFactory factoryInReduce = XFactoryRegistry.instance()
				.currentDefault();
		XTrace preTrace = null;

		@Override
		public void setup(Context context) throws IOException,
				InterruptedException {
			// Timer temptimer1 = new Timer();
			// temptimer1.start();
			Configuration cnfiguration = context.getConfiguration();
			param_xesName = cnfiguration.get("param_xesName");
			param_classify = cnfiguration.get("param_classify");
			log = factoryInReduce.createLog();
			logReplace = factoryInReduce.createLog();
			numOfEvent = 0;
			clock = 0;
			totalEventNum = 0;
			XESConvertorMonitor.instance().numOfEvent = 0;
			// setCaseBranchMap(new Hashtable<String, XTrace>());
			// setup avtivityClassifier
			setupClassifier(log);
			// setup extensions
			setupExtensions(log);
			// setup global attributes
			setupGlobalAttributes(log);
		}

		@Override
		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			String summary = "";
			String keyString = key.toString();
			Iterator valueIterator = values.iterator();
			int removeNum = 0;

			int eventNum = 0;
			boolean correct = true;
			while (valueIterator.hasNext()) {

				String valueString = valueIterator.next().toString();
				// for eventsNum
				if (keyString.equalsIgnoreCase(PanGuConstants.MR_REMOVE)) {
					removeNum++;
					continue;
				}
				// increase the number of events in trace
				eventNum++;
				// System.out.println("valueString in reduce:" + valueString);
				String activityString = "";
				ArrayList<Object> activityIDContentList = new ArrayList<Object>();
				String attributeString = "";
				Hashtable<String, String> attributeIDContentMap = new Hashtable<String, String>();
				// String logPath = "";
				if (valueString.contains(PanGuConstants.MR_TOKEN)) {
					String[] valueDetail = valueString
							.split(PanGuConstants.MR_TOKEN);
					for (int i = 0; i < valueDetail.length; i++) {
						if (valueDetail[i]
								.startsWith(PanGuConstants.MR_ACTIVITY)) {
							activityString = valueDetail[i]
									.substring(PanGuConstants.MR_ACTIVITY
											.length());
							String[] activities = activityString
									.split(PanGuConstants.MR_ACTIVITY_TOKEN);
							for (int j = 0; j < activities.length; j++) {
								activityIDContentList.add(activities[j]);
							}
						} else if (valueDetail[i]
								.startsWith(PanGuConstants.MR_ATTRIBUTE)) {
							attributeString = valueDetail[i]
									.substring(PanGuConstants.MR_ATTRIBUTE
											.length());
							String[] attributes = attributeString
									.split(PanGuConstants.MR_ATTRIBUTE_TOKEN);
							for (int j = 0; j < attributes.length; j++) {

								// String[] attributeDetail =
								// attributes[j].split(":");
								int index = attributes[j].indexOf(":");
								if (index >= 0) {
									String attributeKey = attributes[j]
											.substring(0, index);
									String attributeValue = attributes[j];
									if (attributes[j].substring(index + 1)
											.length() != 0) {
										attributeIDContentMap.put(attributeKey,
												attributeValue);
									} else {
										attributeIDContentMap.put(attributeKey,
												attributeKey + "NA");
									}
								}

							}
						}

					}
				}
				String timeStamp = keyString.substring(keyString.length() - 17,
						keyString.length());
				String caseID = keyString.substring(0, keyString.length() - 17);

				LogBuffer logBuffer = new LogBuffer();
				logBuffer.setActivityIDContentList(activityIDContentList);
				logBuffer.setAttributeIDContentMap(attributeIDContentMap);
				logBuffer.setTimeStamp(timeStamp);
				logBuffer.setCaseIDString(caseID);
				XEvent event = factoryInReduce.createEvent();
				writeEvent(event, logBuffer);
				if (caseID.equals(preCaseID)) {// has the same caseID
					boolean ifTimeOut = false;

					ifTimeOut = XESConvertorMonitor.instance().ifTimeOut(
							preTimestamp, timeStamp);

					if (ifTimeOut) {

						String newCaseID = caseID + BRANCH + (preBranch + 1);
						XTrace traceNew = factoryInReduce.createTrace();

						XAttributeMap traceAttributeMapNew = factoryInReduce
								.createAttributeMap();
						XAttribute traceAttributeID = factoryInReduce
								.createAttributeLiteral(
										XConceptExtension.KEY_NAME, newCaseID,
										null);
						traceAttributeMapNew.put(traceAttributeID.getKey(),
								traceAttributeID);
						traceNew.setAttributes(traceAttributeMapNew);
						traceNew.add(event);
						log.add(traceNew);
						preBranch++;
						preTrace = traceNew;

					} else {// in time zone
						preTrace.add(event);
					}
					// putEventToLog(logBuffer,event);
					preSummary += "\t" + timeStamp;
					preEventNum++;

				} else {// not match the caseID ,it start a new trace

					preSummary = "";
					preSummary += "\t" + timeStamp;
					preEventNum = 0;
					preEventNum++;

					preBranch = -1;
					preCaseID = caseID;

					XTrace traceNew = factoryInReduce.createTrace();
					XAttributeMap traceAttributeMapNew = factoryInReduce
							.createAttributeMap();

					String caseIDString = caseID;
					// add branch to case map
					// getCaseBranchMap().put(caseIDString, traceNew);
					preBranch++;
					caseIDString = caseIDString + BRANCH + preBranch;

					XAttribute traceAttributeNew = factoryInReduce
							.createAttributeLiteral(XConceptExtension.KEY_NAME,
									caseIDString, null);
					traceAttributeMapNew.put(traceAttributeNew.getKey(),
							traceAttributeNew);
					traceNew.setAttributes(traceAttributeMapNew);
					traceNew.add(event);

					// add new trace to log
					log.add(traceNew);
					preTrace = traceNew;
				}

				numOfEvent++;
				clock++;
				totalEventNum++;
				if (clock >= PanGuConstants.MAX_EVENT_NUM) {
					clock = 0;
					correct = splitXLogInReduce(param_xesName, preBranch);
				}
				preTimestamp = timeStamp;// set to current time
			}
			if (keyString.equalsIgnoreCase(PanGuConstants.MR_REMOVE)) {
				context.write(new Text(keyString),
						new Text(String.valueOf(removeNum)));
			}

		}

		// cleanup();

		@Override
		public void cleanup(Reducer.Context context) throws IOException,
				InterruptedException {

			splitFinalLogInReduce(param_xesName, preBranch);

			writeToXES(log, param_xesName, preBranch);

			writeToXES(logReplace, param_xesName, preBranch);

		}
	}

	public static class MyPartition extends Partitioner<Text, Text> {

		@Override
		public int getPartition(Text key, Text value, int numPartitions) {
			// TODO Auto-generated method stub

			String keyString = key.toString();
			if (keyString.equalsIgnoreCase(PanGuConstants.MR_REMOVE)) {
				return Math.abs(keyString.hashCode()) % numPartitions;
			}
			// System.out.println("keyString:" + keyString);
			String prefix;
			String timeStamp = keyString.substring(keyString.length() - 17,
					keyString.length());
			String caseID = keyString.substring(0, keyString.length() - 17);
			prefix = caseID;
			return Math.abs(prefix.hashCode()) % numPartitions;
		}
	}

	public static void main(String[] args) throws Exception {
		System.out.println("Class path in main");
		// PrintClasspath.print();
		Configuration configuration = new Configuration();
		String[] otherArgs = new GenericOptionsParser(configuration, args)
				.getRemainingArgs();
		// configuration.
		Job job = new Job(configuration, "processMining");
		job.setJarByClass(Test.class);
		// the keys are words (strings)
		job.setOutputKeyClass(Text.class);
		// the values are counts (ints)
		job.setOutputValueClass(Text.class);
		job.setPartitionerClass(Test.MyPartition.class);
		job.setMapperClass(Test.Map.class);

		job.setReducerClass(Test.Reduce.class);
		PropertiesReaderServices x1 = PropertiesReaderServicesImpl.instance();
		double x3 = 0.95 * Double.parseDouble(x1.getProperties("hadoopnodes"));
		int rn = (int) x3;
		job.setNumReduceTasks(rn);// set reduce numbers:9*0.95=8.55 9*1.75=15.75
		String email = "";
		// ClassLoader.

		List<String> other_args = new ArrayList<String>();
		try {
			job.getConfiguration().set("param_length",
					String.valueOf(args.length));
		} catch (Exception e) {

		}
		// job.getConfiguration().setBoolean("hadoop.native.lib", false);
		job.getConfiguration().set("tmpjars", "/tmp/libthrift-0.6.jar");

		for (int i = 0; i < args.length; ++i) {
			try {
				if (args[i].startsWith("-m")) {
					// job.setNumMapTasks(Integer.parseInt(args[i].substring(2)));
				} else if (args[i].startsWith(MapReduceComand.COMD_REDUCE)) {
					job.setNumReduceTasks(Integer.parseInt(args[i]
							.substring(MapReduceComand.COMD_REDUCE.length())));
				} else if (args[i].startsWith(MapReduceComand.COMD_VERSION)) {// versionName
					param_version = args[i]
							.substring(MapReduceComand.COMD_VERSION.length());
					job.getConfiguration().set("param_version", param_version);
				} else if (args[i].startsWith(MapReduceComand.COMD_PATTERN)) {// patternName
					param_pattern = args[i]
							.substring(MapReduceComand.COMD_PATTERN.length());
					job.getConfiguration().set("param_pattern", param_pattern);
				} else if (args[i].startsWith(MapReduceComand.COMD_INPUT)) {// inputpath
					param_inputPath = args[i]
							.substring(MapReduceComand.COMD_INPUT.length());
					FileInputFormat.setInputPaths(job, param_inputPath);
				} else if (args[i].startsWith(MapReduceComand.COMD_OUTPUT)) {// outputpath
					param_outputPath = args[i]
							.substring(MapReduceComand.COMD_OUTPUT.length());
					FileOutputFormat.setOutputPath(job, new Path(
							param_outputPath));
				} else if (args[i].startsWith(MapReduceComand.COMD_XESNAME)) {// xesname
					param_xesName = args[i]
							.substring(MapReduceComand.COMD_XESNAME.length());
					job.getConfiguration().set("param_xesName", param_xesName);
				} else if (args[i].startsWith(MapReduceComand.COMD_NIOISEFILE)) {// noisefilename
					param_noiseFile = args[i]
							.substring(MapReduceComand.COMD_NIOISEFILE.length());
					job.getConfiguration().set("param_noiseFile",
							param_noiseFile);
				} else if (args[i].startsWith(MapReduceComand.COMD_CLASSIFY)) {// noisefilename
					param_classify = args[i]
							.substring(MapReduceComand.COMD_CLASSIFY.length());
					job.getConfiguration()
							.set("param_classify", param_classify);
				} else if (args[i].startsWith(MapReduceComand.ue)) {
					// email = args[i].substring(MapReduceComand.ue.length());
				} else if (args[i]
						.startsWith(MapReduceComand.COMD_STARTINGDATE)) {
					param_startingdate = args[i]
							.substring(MapReduceComand.COMD_STARTINGDATE
									.length());
					if (!param_endingdate.equals("")
							&& !param_productlist.equals("")) {
						ParameterConstructor pc = new ParameterConstructor();
						String inputpath = pc.InputPathConstructor(
								param_startingdate, param_endingdate,
								param_productlist);
						FileInputFormat.setInputPaths(job, inputpath);
					}
					other_args.add(args[i]);
				} else if (args[i].startsWith(MapReduceComand.COMD_ENDINGDATE)) {
					param_endingdate = args[i]
							.substring(MapReduceComand.COMD_STARTINGDATE
									.length());
					if (!param_startingdate.equals("")
							&& !param_productlist.equals("")) {
						ParameterConstructor pc = new ParameterConstructor();
						String inputpath = pc.InputPathConstructor(
								param_startingdate, param_endingdate,
								param_productlist);
						FileInputFormat.setInputPaths(job, inputpath);
					}
					other_args.add(args[i]);
				} else if (args[i].startsWith(MapReduceComand.COMD_PRODUCTLIST)) {
					param_productlist = args[i]
							.substring(MapReduceComand.COMD_PRODUCTLIST
									.length());
					if (!param_startingdate.equals("")
							&& !param_endingdate.equals("")) {
						ParameterConstructor pc = new ParameterConstructor();
						String inputpath = pc.InputPathConstructor(
								param_startingdate, param_endingdate,
								param_productlist);
						FileInputFormat.setInputPaths(job, inputpath);
					}
					other_args.add(args[i]);
				} else {
					other_args.add(args[i]);
				}

			} catch (NumberFormatException except) {
				System.out.println("ERROR: Integer expected instead of "
						+ args[i]);

			} catch (ArrayIndexOutOfBoundsException except) {
				System.out.println("ERROR: Required parameter missing from "
						+ args[i - 1]);
			}
		}
		if (job.waitForCompletion(true)) {
			if (!email.equals("")) {
				emails.emailExecute("810586574@qq.com", email, "XES 已经生成成功",
						"您好，你所需要的XES:" + param_xesName
								+ "已经生成，您可以进行后续操作了，：-)!!!", "guofen1988*");
			}
			System.exit(0);
		} else if (!email.equals(""))
			emails.emailExecute(
					"810586574@qq.com",
					email,
					"XES 生成shi",
					"::>_<::，非常抱歉，你所需要的XES:" + param_xesName + "生成失败，请联系管理员!!!",
					"guofen1988*");
		System.exit(1);
	}
	// System.exit(job.waitForCompletion(true) ? 0 : 1);
}
